package com.hoperun.nocserver.common.conf;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.nutz.lang.Strings;
import org.nutz.log.Log;
import org.nutz.log.Logs;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import com.rabbitmq.client.Channel;

/**
 * rabbitMQ配置类
 * 
 * @author zhu_longxiang
 *
 */
@Component
@Configuration
public class RabbitMQConfig {

	private static final Log LOGGER = Logs.get();
	public static final Integer THREAD_POOL_SIZE = 128;
	public static final String EXCHANGE = "noc.server.exchange";
	public static final String ROUTINGKEY = "noc.server.routingkey";
	public static final String RADIO = "noc.server.radio";
	public static final String ADMIN_UI_APP_DISCOVERY = "noc.server.admin.ui.app.discovery";
	public static final String FLEX_DISCOVERY = "noc.server.flex.discovery";
	// 测试队列
	public static final String TEST = "noc.server.test";

	@Autowired
	RabbitProperties rabbitProperties;

	@Bean
	public ConnectionFactory connectionFactory() {
		CachingConnectionFactory connectionFactory = new CachingConnectionFactory(rabbitProperties.getAddresses(),
				rabbitProperties.getPort());
		connectionFactory.setUsername(rabbitProperties.getUsername());
		connectionFactory.setPassword(rabbitProperties.getPassword());
		connectionFactory.setVirtualHost(rabbitProperties.getVirtualHost());
		// 支持发布确认
		connectionFactory.setPublisherConfirms(true);
		connectionFactory.setVirtualHost("/");
		// 支持发布返回
		// connectionFactory.setPublisherReturns(true);
		return connectionFactory;
	}

	@Bean
	@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
	public RabbitTemplate rabbitTemplate() {
		RabbitTemplate template = new RabbitTemplate(connectionFactory());
		template.setMessageConverter(new Jackson2JsonMessageConverter());
		return template;
	}

	/**
	 * 配置交换机
	 * 
	 * @return
	 */
	@Bean
	public DirectExchange defaultExchange() {
		return new DirectExchange(EXCHANGE, true, false);
	}

	/**
	 * 配置广播队列（持久化）
	 * 
	 * @return
	 */
	@Bean
	public Queue radio() {
		return new Queue(RADIO, true);
	}

	/**
	 * 配置admin-ui-app应用程序发现队列（持久化）
	 * 
	 * @return
	 */
	@Bean
	public Queue appDiscovery() {
		return new Queue(ADMIN_UI_APP_DISCOVERY, true);
	}

	/**
	 * 配置flex设备发现队列（持久化）
	 * 
	 * @return
	 */
	@Bean
	public Queue flexDiscovery() {
		return new Queue(FLEX_DISCOVERY, true);
	}

	@Bean
	public Queue test() {
		return new Queue(TEST, true);
	}

	/**
	 * 将广播队列和交换机、路由绑定
	 * 
	 * @return
	 */
	@Bean
	public Binding radioBinding() {
		return BindingBuilder.bind(radio()).to(defaultExchange()).with(RabbitMQConfig.ROUTINGKEY);
	}

	/**
	 * 将admin-ui-app应用程序发现队列和交换机、路由绑定
	 * 
	 * @return
	 */
	@Bean
	public Binding appDiscoveryBinding() {
		return BindingBuilder.bind(appDiscovery()).to(defaultExchange()).with(RabbitMQConfig.ROUTINGKEY);
	}

	/**
	 * 将flex设备发现队列和交换机、路由绑定
	 * 
	 * @return
	 */
	@Bean
	public Binding flexDiscoveryBinding() {
		return BindingBuilder.bind(flexDiscovery()).to(defaultExchange()).with(RabbitMQConfig.ROUTINGKEY);
	}

	@Bean
	public Binding testBinding() {
		return BindingBuilder.bind(test()).to(defaultExchange()).with(RabbitMQConfig.ROUTINGKEY);
	}

	/**
	 * 设置广播队列监听
	 * 
	 * @return
	 */
	@Bean
	public SimpleMessageListenerContainer messageContainer() {
		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
		container.setQueues(radio());
		container.setExposeListenerChannel(true);
		// container.setMaxConcurrentConsumers(200);
		// container.setConcurrentConsumers(4);
		ExecutorService executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
		container.setTaskExecutor(executorService); // 设置确认模式手工确认
		container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
		container.setMessageListener(new ChannelAwareMessageListener() {

			@Override
			public void onMessage(Message message, Channel channel) throws Exception {
				try {
					byte[] body = message.getBody();
					LOGGER.debug("receive msg : " + new String(body));
					// TODO 业务处理
				} catch (Exception e) {
					LOGGER.error("处理消息异常！", e);
				} finally {
					channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
				}
			}
		});

		return container;
	}

	/**
	 * 设置admin-ui-app应用程序发现队列监听
	 * 
	 * @return
	 */
	@Bean
	public SimpleMessageListenerContainer appContainer() {
		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
		container.setQueues(appDiscovery());
		container.setExposeListenerChannel(true);
		// container.setMaxConcurrentConsumers(200);
		// container.setConcurrentConsumers(4);
		ExecutorService executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
		container.setTaskExecutor(executorService); // 设置确认模式手工确认
		container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
		container.setMessageListener(new ChannelAwareMessageListener() {

			@Override
			public void onMessage(Message message, Channel channel) throws Exception {
				try {
					byte[] body = message.getBody();
					LOGGER.debug("receive msg : " + new String(body));
					// TODO 可以做设置创建相应处理队列和监听操作
				} catch (Exception e) {
					LOGGER.error("处理消息异常！", e);
				} finally {
					channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
				}
			}
		});

		return container;
	}

	/**
	 * 设置flex设备发现队列监听
	 * 
	 * @return
	 */
	@Bean
	public SimpleMessageListenerContainer flexContainer() {
		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
		container.setQueues(flexDiscovery());
		container.setExposeListenerChannel(true);
		// container.setMaxConcurrentConsumers(200);
		// container.setConcurrentConsumers(4);
		ExecutorService executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
		container.setTaskExecutor(executorService); // 设置确认模式手工确认
		container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
		container.setMessageListener(new ChannelAwareMessageListener() {

			@Override
			public void onMessage(Message message, Channel channel) throws Exception {
				try {
					byte[] body = message.getBody();
					LOGGER.debug("receive msg : " + new String(body));
					// TODO 可以做设置创建相应处理队列和监听操作
				} catch (Exception e) {
					LOGGER.error("处理消息异常！", e);
				} finally {
					channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
				}
			}
		});

		return container;
	}

	@Bean
	public SimpleMessageListenerContainer testContainer() {
		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
		container.setQueues(flexDiscovery());
		container.setExposeListenerChannel(true);
		// container.setMaxConcurrentConsumers(200);
		// container.setConcurrentConsumers(4);
		ExecutorService executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
		container.setTaskExecutor(executorService); // 设置确认模式手工确认
		container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
		container.setMessageListener(new ChannelAwareMessageListener() {

			@Override
			public void onMessage(Message message, Channel channel) throws Exception {
				try {
					String msg = new String(message.getBody());
					LOGGER.debug("receive msg : " + msg);
					if (!Strings.isBlank(msg)) {
						// TODO 测试处理

					}
				} catch (Exception e) {
					LOGGER.error("处理消息异常！", e);
				} finally {
					channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
				}
			}
		});

		return container;
	}

}
